package com.taurus.permanent.normal;

import java.io.IOException;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.taurus.core.util.Utils;
import com.taurus.permanent.core.BaseCoreService;
import com.taurus.permanent.core.BitSwarmEngine;
import com.taurus.permanent.io.IOHandler;

public class DatagramReader extends BaseCoreService implements Runnable {
	private final BitSwarmEngine engine;

	private final Logger logger;

	private final Thread readerThread;

	private Selector udpSelector;

	private IOHandler ioHandler;

	private volatile boolean isActive = false;

	private volatile long readBytes = 0L;

	public DatagramReader() {
		this.engine = BitSwarmEngine.getInstance();
		this.logger = LoggerFactory.getLogger(getClass());
		try {
			this.udpSelector = Selector.open();
			this.logger.info("UDP Selector opened");
		} catch (IOException e) {
			this.logger.error("Failed opening TCP Selector: " + e.toString());
			e.printStackTrace();
		}
		this.readerThread = new Thread(this, "DatagramReader");
	}

	public void init(Object o) {
		if (this.isActive)
			throw new IllegalArgumentException("Object is already initialized. Destroy it first!");
		this.isActive = true;
		this.readerThread.start();
		this.logger.info("DatagramReader started");
	}

	public void destroy(Object o) {
		this.isActive = false;
		try {
			Thread.sleep(500L);
			this.udpSelector.close();
		} catch (Exception e) {
			this.logger.error("Error when shutting down UDP Selector: ", e);
		}
	}

	public void run() {
		ByteBuffer readBuffer = Utils.allocateBuffer(this.engine.getConfig().maxReadBufferSize,
				this.engine.getConfig().readBufferType);
		while (this.isActive) {
			try {
				readIncomingDatagrams(readBuffer);
			} catch (Throwable t) {
				logger.error("Problems in DatagramReader main loop: ", t);
			}
		}
		this.logger.info("DatagramReader thread(s) shutting down.");
	}

	private void readIncomingDatagrams(ByteBuffer readBuffer) {
		DatagramChannel chan = null;
		SelectionKey key = null;
		try {
			this.udpSelector.select();
			Iterator<SelectionKey> selectedKeys = this.udpSelector.selectedKeys().iterator();
			while (selectedKeys.hasNext()) {
				try {
					key = selectedKeys.next();
					selectedKeys.remove();
					if (!key.isValid())
						continue;
					if (key.isReadable()) {
						readBuffer.clear();
						chan = (DatagramChannel) key.channel();
						readPacket(chan, readBuffer);
					}
				} catch (IOException e) {
					this.logger.error("Problem reading UDP Packet, from: %s, Error: %s", chan.toString(), e.toString());
				}
			}
		} catch (ClosedSelectorException e) {
			this.logger.debug("Selector is closed!");
		} catch (CancelledKeyException cancelledKeyException) {
		} catch (IOException ioe) {
			this.logger.error("Datagram selection IOError: ", ioe);
		} catch (Exception err) {
			this.logger.error("Generic reading/selection error: ", err);
		}
	}

	private void readPacket(DatagramChannel chan, ByteBuffer readBuffer) throws IOException {
		long byteCount = 0L;
		SocketAddress address = chan.receive(readBuffer);
		if (address != null) {
			byteCount = readBuffer.position();
			if (byteCount > 0L) {
				this.readBytes += byteCount;
				readBuffer.flip();
				byte[] binaryData = new byte[readBuffer.limit()];
				readBuffer.get(binaryData);
				this.ioHandler.onDataRead(chan, address, binaryData);
			}
		} else {
			this.logger.info("Could not read any data from DatagramChannel: " + chan);
		}
	}

	public IOHandler getIOHandler() {
		return this.ioHandler;
	}

	public long getReadBytes() {
		return this.readBytes;
	}

	public long getReadPackets() {
		return this.ioHandler.getReadPackets();
	}

	public void setIoHandler(IOHandler handler) {
		if (handler == null)
			throw new IllegalStateException("IOHandler is already set!");
		this.ioHandler = handler;
	}

	public Selector getSelector() {
		return this.udpSelector;
	}
}
